package cn.doitedu.demo4;

import cn.doitedu.beans.UserAction;
import cn.doitedu.common.utils.CommonUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/8
 * @Desc: 学大数据，上多易教育
 * <p>
 * 模型2规则示例：
 * 静态画像条件：
 * age>10,gender=male
 * 动态画像统计条件：
 * X>=3次
 * 触发条件：
 * Y行为
 * <p>
 * 参数模板：
{
"rule_id":"rule-m02-004",
"static_profile_conditions":[
{
"tag_name":"age",
"compare_type":">",
"compare_value":10
},
{
"tag_name":"gender",
"compare_type":"=",
"compare_value":"male"
}
],
"realtime_statistic_conditions":{
"event_id":"X",
"min_count":3
},
"trigger_event_id":"Y"
}
 **/
public class RuleModelCalculator_02 implements RuleCalculator {

    JSONObject paramObject;
    private String ruleId;
    Table table;
    ValueState<Integer> cntState;

    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson) throws IOException {

        this.paramObject = JSON.parseObject(ruleParamJson);
        this.ruleId = paramObject.getString("rule_id");

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection connection = ConnectionFactory.createConnection(conf);
        this.table = connection.getTable(TableName.valueOf("user_profile_tags"));

        this.cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-cnt", Integer.class));

        this.message = new JSONObject();
        this.message.put("rule_id", ruleId);

    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {

        String eventId = userAction.getEvent_id();

        /* *
         * 动态画像统计
         */
        // 从规则参数中获取动态画像统计的目标事件
        JSONObject obj = paramObject.getJSONObject("realtime_statistic_conditions");
        String targetEventId = obj.getString("event_id");
        int targetMinCount = obj.getIntValue("min_count");

        if (eventId.equals(targetEventId)) {
            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);
        }


        /* *
         * 触发判断
         */
        String triggerEventId = paramObject.getString("trigger_event_id");
        if (eventId.equals(triggerEventId)) {
            // 判断动态统计条件是否已满足
            if (cntState.value() != null && cntState.value() >= targetMinCount){
                // 继续判断静态画像条件是否满足

                // 指定行键，创建一个hbase查询用的get条件对象
                Get get = new Get(Bytes.toBytes(userAction.getUser_id()));

                // 遍历静态画像条件，将每个要查询的标签，放入get条件中
                JSONArray array = paramObject.getJSONArray("static_profile_conditions");
                for(int i=0;i<array.size();i++){
                    JSONObject condition_i = array.getJSONObject(i);
                    String tagName = condition_i.getString("tag_name");
                    // 将本条件的标签名（字段名），放入get条件中
                    get.addColumn("f".getBytes(),tagName.getBytes());
                }

                // 用table客户端根据组装好的get条件进行查询
                Result result = table.get(get);

                boolean compareOk = false;
                for(int i=0;i<array.size();i++){

                    JSONObject condition_i = array.getJSONObject(i);
                    String tagName = condition_i.getString("tag_name");

                    // >   >=   <
                    String compareType = condition_i.getString("compare_type");
                    String compareValue = condition_i.getString("compare_value");

                    byte[] valueBytes = result.getValue("f".getBytes(), tagName.getBytes());
                    String resultTagValue = Bytes.toString(valueBytes);

                    // 比较参数中的值和  查询到的值
                    compareOk = CommonUtils.compare(compareValue, resultTagValue, compareType);

                    // 但凡有一个条件不满足，直接return，calc方法结束
                    if(! compareOk) return;

                }


                // 如果代码走到这里，说明所有静态画像条件都满足，则发消息
                message.put("user_id",userAction.getUser_id());
                message.put("hit_time",userAction.getAction_time());

                out.collect(message.toJSONString());

            }

        }


    }

    @Override
    public String ruleId() {
        return ruleId;
    }
}
